/*
 * @Author: caiwanli 651943559@qq.com
 * @Date: 2023-12-14 12:47:37
 * @LastEditors: caiwanli 651943559@qq.com
 * @LastEditTime: 2024-01-17 16:17:26
 * @FilePath: /task_sche/src/binder/bind_select.cpp
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置
 * 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
#include "binder.h"
#include "bound_agg_call.h"
#include "bound_base_table_ref.h"
#include "bound_binary_op.h"
#include "bound_column_ref.h"
#include "bound_constant.h"
#include "bound_cross_product_ref.h"
#include "bound_join_ref.h"
#include "bound_star.h"
#include "bound_unary_op.h"
#include "catalog.hpp"
#include "exception.h"
#include "nodes/nodes.hpp"
#include "nodes/parsenodes.hpp"
#include "nodes/primnodes.hpp"
#include "pg_definitions.hpp"
#include "postgres_parser.hpp"

#include <cstdio>
#include <memory>

namespace DaseX {

auto Binder::BindRangeVar(duckdb_libpgquery::PGRangeVar *table_ref)
    -> std::unique_ptr<BoundTableRef> {
  // TODO: "with"
  // realation name ,from (from)
  return BindBaseTableRef(table_ref->relname, std::nullopt);
}

auto Binder::BindJoin(duckdb_libpgquery::PGJoinExpr *root)
    -> std::unique_ptr<BoundTableRef> {
  JoinType join_type;
  switch (root->jointype) {
  case duckdb_libpgquery::PG_JOIN_INNER: {
    join_type = JoinType::INNER;
    break;
  }
  case duckdb_libpgquery::PG_JOIN_LEFT: {
    join_type = JoinType::LEFT;
    break;
  }
  case duckdb_libpgquery::PG_JOIN_FULL: {
    join_type = JoinType::OUTER;
    break;
  }
  case duckdb_libpgquery::PG_JOIN_RIGHT: {
    join_type = JoinType::RIGHT;
    break;
  }
  default: {
    throw DaseX::Exception("Join type {} not supported");
  }
  }
  auto left_table = BindTableRef(root->larg);
  auto right_table = BindTableRef(root->rarg);
  auto join_ref = std::make_unique<BoundJoinRef>(
      join_type, std::move(left_table), std::move(right_table), nullptr);
  scope_ = join_ref.get();
  auto condition = BindExpression(root->quals);
  join_ref->condition_ = std::move(condition);
  return join_ref;
}

auto Binder::BindBaseTableRef(std::string table_name,
                              std::optional<std::string> alias)
    -> std::unique_ptr<BoundBaseTableRef> {
  auto table = catalog_.tables.at(table_name);

  return std::make_unique<BoundBaseTableRef>(std::move(table_name),
                                             std::move(alias), table->schema);
}

auto Binder::BindTableRef(duckdb_libpgquery::PGNode *node)
    -> std::unique_ptr<BoundTableRef> {
  switch (node->type) {
  case duckdb_libpgquery::T_PGRangeVar: {
    return BindRangeVar(
        reinterpret_cast<duckdb_libpgquery::PGRangeVar *>(node));
  }
  case duckdb_libpgquery::T_PGJoinExpr: {
    return BindJoin(reinterpret_cast<duckdb_libpgquery::PGJoinExpr *>(node));
  }
  case duckdb_libpgquery::T_PGRangeSubselect: {
    throw DaseX::Exception("unsupported node type: {}");
    //
  }
  default:
    throw DaseX::Exception("unsupported node type: {}");
  }
}

auto Binder::BindFrom(duckdb_libpgquery::PGList *list)
    -> std::unique_ptr<BoundTableRef> {

  if (list == nullptr) {
    return std::make_unique<BoundTableRef>(TableReferenceType::EMPTY);
  }
  if (list->length > 1) {
    // select .. from x,y

    // first node.
    printf("The first node\n");
    auto c = list->head;
    auto lnode =
        reinterpret_cast<duckdb_libpgquery::PGNode *>(c->data.ptr_value);
    auto ltable = BindTableRef(lnode);
    c = c->next;

    // second node.
    printf("The second node\n");
    auto rnode =
        reinterpret_cast<duckdb_libpgquery::PGNode *>(c->data.ptr_value);
    auto rtable = BindTableRef(rnode);
    c = c->next;

    auto result = std::make_unique<BoundCrossProductRef>(std::move(ltable),
                                                         std::move(rtable));

    for (; c != nullptr; c = c->next) {
      auto node =
          reinterpret_cast<duckdb_libpgquery::PGNode *>(c->data.ptr_value);
      auto table = BindTableRef(node);
      result = std::make_unique<BoundCrossProductRef>(std::move(result),
                                                      std::move(table));
    }

    return result;
  }

  auto node =
      reinterpret_cast<duckdb_libpgquery::PGNode *>(list->head->data.ptr_value);
  return BindTableRef(node);
}

auto Binder::GetAllColumns(const BoundTableRef &scope)
    -> std::vector<std::unique_ptr<BoundExpression>> {
  switch (scope.type_) {
  case TableReferenceType::BASE_TABLE: {
    const auto &base_table_ref = dynamic_cast<const BoundBaseTableRef &>(scope);
    auto bound_table_name = base_table_ref.GetBoundTableName();
    const auto &schema = base_table_ref.schema_;
    auto columns = std::vector<std::unique_ptr<BoundExpression>>{};

    for (const auto &column : schema.fields()) {
      columns.push_back(std::make_unique<BoundColumnRef>(
          std::vector{bound_table_name, column->name()}));
    }
    return columns;
  }
  case TableReferenceType::CROSS_PRODUCT: {
    const auto &cross_product_ref =
        dynamic_cast<const BoundCrossProductRef &>(scope);
    auto columns = GetAllColumns(*cross_product_ref.left_);
    auto append_columns = GetAllColumns(*cross_product_ref.right_);
    std::copy(std::make_move_iterator(append_columns.begin()),
              std::make_move_iterator(append_columns.end()),
              std::back_inserter(columns));
    // to del
    for (auto &column : columns) {
      if (column->type_ == ExpressionType::COLUMN_REF) {
        auto &column_ref = dynamic_cast<BoundColumnRef &>(*column);
        auto temp_string = column_ref.ToString();
        // printf("cross cols :%s ",column_ref.col_name_[1].c_str());
        printf("cross cols :%s ", temp_string.c_str());
      }
    }
    printf("\n");
    return columns;
  }
  case TableReferenceType::JOIN: {
    const auto &join_ref = dynamic_cast<const BoundJoinRef &>(scope);
    auto columns = GetAllColumns(*join_ref.left_);
    auto append_columns = GetAllColumns(*join_ref.right_);
    std::copy(std::make_move_iterator(append_columns.begin()),
              std::make_move_iterator(append_columns.end()),
              std::back_inserter(columns));
    return columns;
  }

  default:
    throw DaseX::Exception(
        "select * cannot be used with this TableReferenceType");
  }
}

static auto ResolveColumnRefFromSchema(const arrow::Schema &schema,
                                       const std::vector<std::string> &col_name)
    -> std::unique_ptr<BoundColumnRef> {
  if (col_name.size() != 1) {
    return nullptr;
  }
  std::unique_ptr<BoundColumnRef> column_ref = nullptr;
  for (const auto &column : schema.fields()) {
    if (column->name() == col_name[0]) {
      column_ref =
          std::make_unique<BoundColumnRef>(std::vector{column->name()});
    }
  }
  return column_ref;
}

auto Binder::ResolveColumnRefFromBaseTableRef(
    const BoundBaseTableRef &table_ref,
    const std::vector<std::string> &col_name)
    -> std::unique_ptr<BoundColumnRef> {
  auto bound_table_name = table_ref.GetBoundTableName();
  std::unique_ptr<BoundColumnRef> direct_resolved_expr =
      BoundColumnRef::Prepend(
          ResolveColumnRefFromSchema(table_ref.schema_, col_name),
          bound_table_name);

  std::unique_ptr<BoundColumnRef> strip_resolved_expr = nullptr;

  if (col_name.size() > 1) {
    if (col_name[0] == bound_table_name) {
      auto strip_column_name = col_name;
      strip_column_name.erase(strip_column_name.begin());
      auto x = ResolveColumnRefFromSchema(table_ref.schema_, strip_column_name);
      strip_resolved_expr =
          BoundColumnRef::Prepend(std::move(x), bound_table_name);
    }
  }

  if (strip_resolved_expr != nullptr) {
    return strip_resolved_expr;
  }
  return direct_resolved_expr;
}

auto Binder::ResolveColumnInternal(const BoundTableRef &table_ref,
                                   const std::vector<std::string> &col_name)
    -> std::unique_ptr<BoundExpression> {
  switch (table_ref.type_) {
  case TableReferenceType::BASE_TABLE: {
    const auto &base_table_ref =
        dynamic_cast<const BoundBaseTableRef &>(table_ref);
    return ResolveColumnRefFromBaseTableRef(base_table_ref, col_name);
  }
  case TableReferenceType::CROSS_PRODUCT: {
    const auto &cross_product_ref =
        dynamic_cast<const BoundCrossProductRef &>(table_ref);
    auto left_column =
        ResolveColumnInternal(*cross_product_ref.left_, col_name);
    auto right_column =
        ResolveColumnInternal(*cross_product_ref.right_, col_name);

    if (left_column != nullptr) {
      return left_column;
    }
    return right_column;
  }
  case TableReferenceType::JOIN: {
    const auto &join_ref = dynamic_cast<const BoundJoinRef &>(table_ref);
    auto left_column = ResolveColumnInternal(*join_ref.left_, col_name);
    auto right_column = ResolveColumnInternal(*join_ref.right_, col_name);

    if (left_column != nullptr) {
      return left_column;
    }
    return right_column;
  }

  }
}

auto Binder::ResolveColumn(const DaseX::BoundTableRef &scope,
                           const std::vector<std::string> &col_name)
    -> std::unique_ptr<DaseX::BoundExpression> {
  auto expr = ResolveColumnInternal(scope, col_name);
  return expr;
}

auto Binder::BindColumnRef(duckdb_libpgquery::PGColumnRef *node)
    -> std::unique_ptr<BoundExpression> {
  auto fields = node->fields;
  auto head_node =
      static_cast<duckdb_libpgquery::PGNode *>(fields->head->data.ptr_value);
  switch (head_node->type) {
  case duckdb_libpgquery::T_PGString: {
    std::vector<std::string> column_names;
    for (auto node = fields->head; node != nullptr; node = node->next) {
      column_names.emplace_back(
          reinterpret_cast<duckdb_libpgquery::PGValue *>(node->data.ptr_value)
              ->val.str);
    }
    return ResolveColumn(*scope_, column_names);
  }
  case duckdb_libpgquery::T_PGAStar: {
    printf("bind Astar\n");
    return BindStar(reinterpret_cast<duckdb_libpgquery::PGAStar *>(head_node));
  }
  }
}

auto Binder::BindStar(duckdb_libpgquery::PGAStar *node)
    -> std::unique_ptr<BoundExpression> {
  return std::make_unique<BoundStar>();
}

auto Binder::BindResTarget(duckdb_libpgquery::PGResTarget *root)
    -> std::unique_ptr<BoundExpression> {
  auto expr = BindExpression(root->val);
  if (!expr) {
    return nullptr;
  }
  if (root->name != nullptr) {
    // TODO:别名处理
  }
  return expr;
}
auto Binder::BindAExpr(duckdb_libpgquery::PGAExpr *root)
    -> std::unique_ptr<BoundExpression> {
  auto name = std::string((reinterpret_cast<duckdb_libpgquery::PGValue *>(
                               root->name->head->data.ptr_value))
                              ->val.str);
  // printf("Expr name : %s \n",name.c_str());
  if (root->kind != duckdb_libpgquery::PG_AEXPR_OP) {
    throw DaseX::Exception("unsupported op in AExpr");
  }

  std::unique_ptr<BoundExpression> left_expr = nullptr;
  std::unique_ptr<BoundExpression> right_expr = nullptr;

  if (root->lexpr != nullptr) {
    left_expr = BindExpression(root->lexpr);
  }
  if (root->rexpr != nullptr) {
    right_expr = BindExpression(root->rexpr);
  }


  // support > = <, but how to sup in/not in
  if (left_expr && right_expr) {
    return std::make_unique<BoundBinaryOp>(name, std::move(left_expr),
                                           std::move(right_expr));
  }
  // if (!left_expr && right_expr) {
  //   return std::make_unique<BoundUnaryOp>(name, std::move(right_expr));
  // }
  throw DaseX::Exception("unsupported AExpr: left == null while right != null");
}

auto Binder::BindExpression(duckdb_libpgquery::PGNode *node)
    -> std::unique_ptr<BoundExpression> {
  switch (node->type) {
  case duckdb_libpgquery::T_PGColumnRef:
    return BindColumnRef(
        reinterpret_cast<duckdb_libpgquery::PGColumnRef *>(node));
  case duckdb_libpgquery::T_PGAConst:
    return BindConstant(reinterpret_cast<duckdb_libpgquery::PGAConst *>(node));
  case duckdb_libpgquery::T_PGResTarget:
    return BindResTarget(
        reinterpret_cast<duckdb_libpgquery::PGResTarget *>(node));
  case duckdb_libpgquery::T_PGAStar:
    return BindStar(reinterpret_cast<duckdb_libpgquery::PGAStar *>(node));
  case duckdb_libpgquery::T_PGFuncCall:
    return BindFuncCall(
        reinterpret_cast<duckdb_libpgquery::PGFuncCall *>(node));
  case duckdb_libpgquery::T_PGAExpr:
    return BindAExpr(reinterpret_cast<duckdb_libpgquery::PGAExpr *>(node));
  case duckdb_libpgquery::T_PGBoolExpr:
    return BindBoolExpr(
        reinterpret_cast<duckdb_libpgquery::PGBoolExpr *>(node));
  default:
    break;
  }
}

auto Binder::BindConstant(duckdb_libpgquery::PGAConst *node)
    -> std::unique_ptr<BoundExpression> {

  const auto &val = node->val;
  switch (val.type) {
  case duckdb_libpgquery::T_PGInteger: {
    return std::make_unique<BoundConstant>(static_cast<int32_t>(val.val.ival));
  }
  case duckdb_libpgquery::T_PGString: {
  }
  case duckdb_libpgquery::T_PGNull: {
    DaseX::Exception("TODO T_PGString\n");
  }
  default:
    break;
  }
}

auto Binder::BindBoolExpr(duckdb_libpgquery::PGBoolExpr *root)
    -> std::unique_ptr<BoundExpression> {
  switch (root->boolop) {
  case duckdb_libpgquery::PG_AND_EXPR:
  case duckdb_libpgquery::PG_OR_EXPR: {
    std::string op_name;
    if (root->boolop == duckdb_libpgquery::PG_AND_EXPR) {
      op_name = "and";
    } else if (root->boolop == duckdb_libpgquery::PG_OR_EXPR) {
      op_name = "or";
    }

    auto exprs = BindExpressionList(root->args);
    if (exprs.size() <= 1) {
      if (root->boolop == duckdb_libpgquery::PG_AND_EXPR) {
        throw DaseX::Exception("AND should have at least 1 arg");
      }
      throw DaseX::Exception("OR should have at least 1 arg");
    }
    auto expr = std::make_unique<BoundBinaryOp>(op_name, std::move(exprs[0]),
                                                std::move(exprs[1]));
    for (size_t i = 2; i < exprs.size(); i++) {
      expr = std::make_unique<BoundBinaryOp>(op_name, std::move(expr),
                                             std::move(exprs[i]));
    }
    return expr;
  }
  case duckdb_libpgquery::PG_NOT_EXPR: {
    auto exprs = BindExpressionList(root->args);
    if (exprs.size() != 1) {
      throw DaseX::Exception("NOT should have 1 arg");
    }
    return std::make_unique<BoundUnaryOp>("not", std::move(exprs[0]));
  }
  }
}

/*
 * TODO values in insert statement
 */
auto Binder::BindExpressionList(duckdb_libpgquery::PGList *list)
    -> std::vector<std::unique_ptr<BoundExpression>> {
  std::vector<std::unique_ptr<BoundExpression>> exprs;
  auto select_list = std::vector<std::unique_ptr<BoundExpression>>{};

  for (auto node = list->head; node != nullptr; node = node->next) {
    auto target =
        reinterpret_cast<duckdb_libpgquery::PGNode *>(node->data.ptr_value);

    auto expr = BindExpression(target);

    select_list.push_back(std::move(expr));
  }

  return select_list;
}

auto Binder::BindFuncCall(duckdb_libpgquery::PGFuncCall *root)
    -> std::unique_ptr<BoundExpression> {
  auto name = root->funcname;
  auto function_char_name =
      reinterpret_cast<duckdb_libpgquery::PGValue *>(name->head->data.ptr_value)
          ->val.str;
  std::string function_name = function_char_name;
  std::vector<std::unique_ptr<BoundExpression>> children;
  if (root->args != nullptr) {
    for (auto node = root->args->head; node != nullptr; node = node->next) {
      auto child_expr = BindExpression(
          static_cast<duckdb_libpgquery::PGNode *>(node->data.ptr_value));
      children.push_back(std::move(child_expr));
    }
  }
  printf("FuncCall func name is: %s", function_name.c_str());

  if (function_name == "min" || function_name == "max" ||
      function_name == "first" || function_name == "last" ||
      function_name == "sum" || function_name == "count") {
    if (function_name == "count" && children.empty()) {
      function_name = "count_star";
    }

    // Bind agg call.
    return std::make_unique<BoundAggCall>(function_name, root->agg_distinct,
                                          std::move(children));
  }
  // why next line cant be used
  // throw DaseX::Exception("unsupported func call {}");
}

auto Binder::BindSelectList(duckdb_libpgquery::PGList *list)
    -> std::vector<std::unique_ptr<BoundExpression>> {
  std::vector<std::unique_ptr<BoundExpression>> exprs;
  auto select_list = std::vector<std::unique_ptr<BoundExpression>>{};
  bool is_select_star = false;

  for (auto node = list->head; node != nullptr; node = node->next) {
    auto target =
        reinterpret_cast<duckdb_libpgquery::PGNode *>(node->data.ptr_value);

    auto expr = BindExpression(target);

    if (expr->type_ == ExpressionType::STAR) {
      // "SELECT *".
      select_list = GetAllColumns(*scope_);
      is_select_star = true;
    } else {
      select_list.push_back(std::move(expr));
    }
  }
  return select_list;
}

auto Binder::BindWhere(duckdb_libpgquery::PGNode *root)
    -> std::unique_ptr<BoundExpression> {
  return BindExpression(root);
}

auto Binder::BindGroupBy(duckdb_libpgquery::PGList *list)
    -> std::vector<std::unique_ptr<BoundExpression>> {
  return BindExpressionList(list);
}

auto Binder::BindHaving(duckdb_libpgquery::PGNode *root)
    -> std::unique_ptr<BoundExpression> {
  return BindExpression(root);
}

auto Binder::BindSort(duckdb_libpgquery::PGList *list)
    -> std::vector<std::unique_ptr<BoundOrderBy>> {
  auto order_by = std::vector<std::unique_ptr<BoundOrderBy>>{};

  for (auto node = list->head; node != nullptr; node = node->next) {
    auto temp =
        reinterpret_cast<duckdb_libpgquery::PGNode *>(node->data.ptr_value);
    if (temp->type == duckdb_libpgquery::T_PGSortBy) {
      OrderByType type;
      auto sort = reinterpret_cast<duckdb_libpgquery::PGSortBy *>(temp);
      auto target = sort->node;
      if (sort->sortby_dir == duckdb_libpgquery::PG_SORTBY_DEFAULT) {
        type = OrderByType::DEFAULT;
      } else if (sort->sortby_dir == duckdb_libpgquery::PG_SORTBY_ASC) {
        type = OrderByType::ASC;
      } else if (sort->sortby_dir == duckdb_libpgquery::PG_SORTBY_DESC) {
        type = OrderByType::DESC;
      } else {
        throw NotImplementedException("unimplemented order by type");
      }
      auto order_expression = BindExpression(target);
      order_by.emplace_back(
          std::make_unique<BoundOrderBy>(type, std::move(order_expression)));
    } else {
      throw NotImplementedException("unsupported order by node");
    }
  }
  return order_by;
}

auto Binder::BindSelect(duckdb_libpgquery::PGSelectStmt *pg_stmt)
    -> std::unique_ptr<SelectStatement> {

  // Bind FROM clause.
  auto table = BindFrom(pg_stmt->fromClause);
  scope_ = table.get();

  // Bind SELECT list.
  if (pg_stmt->targetList == nullptr) {
    throw DaseX::Exception("no select list");
  }
  // TODO agg call failed, fix name
  auto select_list = BindSelectList(pg_stmt->targetList);

  // Bind WHERE clause.
  auto where = std::make_unique<BoundExpression>();
  if (pg_stmt->whereClause != nullptr) {
    printf("where\n");
    where = BindWhere(pg_stmt->whereClause);
  }

  // Bind GROUP BY clause.
  auto group_by = std::vector<std::unique_ptr<BoundExpression>>{};
  if (pg_stmt->groupClause != nullptr) {
    group_by = BindGroupBy(pg_stmt->groupClause);
  }

  // Bind HAVING clause.
  auto having = std::make_unique<BoundExpression>();
  if (pg_stmt->havingClause != nullptr) {
    having = BindHaving(pg_stmt->havingClause);
  }

  // auto limit_count = std::make_unique<BoundExpression>();
  // // Bind LIMIT clause.
  // if (pg_stmt->limitCount != nullptr) {
  //   limit_count = BindLimitCount(pg_stmt->limitCount);
  // }

  // // Bind OFFSET clause.
  // auto limit_offset = std::make_unique<BoundExpression>();
  // if (pg_stmt->limitOffset != nullptr) {
  //   limit_offset = BindLimitOffset(pg_stmt->limitOffset);
  // }

  // Bind ORDER BY clause.
  auto sort = std::vector<std::unique_ptr<BoundOrderBy>>{};
  if (pg_stmt->sortClause != nullptr) {
    sort = BindSort(pg_stmt->sortClause);
  }

  return std::make_unique<SelectStatement>(
      std::move(table), std::move(select_list), std::move(where),
      std::move(group_by), std::move(having), std::move(sort));
}

} // namespace DaseX